package cn.com.greatwall.kafka.util;

import java.io.Closeable;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

/**
 * @Author herw
 * @Time 2021-05-20 15:24:59
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
public class Consumers<K, V> implements Closeable {
    
    private KafkaConsumer<K, V> consumer;
    
    /**
     *  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperties.getBootstrap());
     *   props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroup());
     *   props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
     *   props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
     *   props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
     */
    public Consumers(Properties properties) {
        this.consumer = new KafkaConsumer(properties);
    }
    
    public KafkaConsumer<K, V> subscribe(String topicName) {
        this.consumer.subscribe(Collections.singleton(topicName));
        return this.consumer;
    }
    
    public KafkaConsumer<K, V> assign(TopicPartition topicPartition) {
        this.consumer.assign(Collections.singleton(topicPartition));
        return this.consumer;
    }
    
    public KafkaConsumer<K, V> seek(TopicPartition topicPartition, Long offset) {
        this.consumer.seek(topicPartition, offset);
        return this.consumer;
    }
    
    public ConsumerRecords<K, V> poll(Duration duration) {
        return this.consumer.poll(duration);
    }
    
    public void commit() {
        this.consumer.commitSync();
    }
    
    public void commitByPartition(Map<TopicPartition, OffsetAndMetadata> offsets) {
        this.consumer.commitSync(offsets);
    }
    
    @Override
    public void close() throws IOException {
        this.consumer.close();
    }
}
